package org.gorson.sink;


import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.gorson.BO.AssetDetailBO;

import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class SinkToMysqlPersonAsset extends RichSinkFunction<AssetDetailBO> {
    private static PreparedStatement ps = null;
    private static Connection connection = null;
    private Integer batchNum = 0;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        String sql = "insert into person_asset (trans_date, pay_code, trans_amt, trans_summary, trans_firm, trans_status) values(?, ?, ?, ?, ?, ?)";
        connection = getConnection();
        System.out.println(connection.hashCode());
        ps = connection.prepareStatement(sql);
    }

    @Override
    public void close() throws Exception {
        super.close();
        //关闭连接
        ps.close();
        connection.close();
    }

    @Override
    public void invoke(AssetDetailBO assetDetail, Context context) throws Exception {
        ps.setString(1, assetDetail.getTransDate());
        ps.setString(2, assetDetail.getPayCode());
        ps.setString(3, assetDetail.getTransAmt());
        ps.setString(4, assetDetail.getTransSummary());
        ps.setString(5, assetDetail.getTransFirm());
        ps.setString(6, assetDetail.getTransStatus());
        ps.execute();
    }

    private static Connection getConnection() throws Exception {
//        ParameterTool parameterTool = ParameterTool.fromPropertiesFile(CommonConstance.propertiesPath);
        Connection con = null;
        try {
            Class.forName("com.mysql.jdbc.Driver");
//            Class.forName("com.mysql.cj.jdbc.Driver");
            //不使用连接池
            con = DriverManager.getConnection(
                    "jdbc:mysql://127.0.0.1:3306/flink?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC&useSSL=false&rewriteBatchedStatements=true",
                    "root",
                    "123456");
            //使用druid连接池
//            con = dataSource.getConnection();
//            if (connection == null) {
//                synchronized (DruidDataSource.class) {
//                    if (connection == null) {
//                        connection = DriverManager.getConnection(parameterTool.get("DATA_BASE_ADDRESS") + parameterTool.get("DATA_CONFIG"), parameterTool.get("DATA_BASE_USER_NAME"), parameterTool.get("DATA_BASE_USER_PASSWORD"));
//                        return connection;
//                    }
//                }
//            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return con;
    }
}
